package com.jwater.core.manager.master;

import com.jwater.core.common.S;
import com.jwater.core.manager.common.DataKeys;
import com.jwater.core.manager.common.ManagerException;
import com.jwater.core.manager.common.MsgKeys;
import com.jwater.core.manager.common.conf.CoreConfParse;
import com.jwater.core.manager.common.conf.JobConfParse;
import com.jwater.core.manager.common.conf.WorkerConfParse;
import com.jwater.core.manager.common.model.*;
import com.jwater.core.manager.job.JobScheduler;
import com.jwater.core.manager.master.check.CoreConfCheck;
import com.jwater.core.manager.master.check.JobCheck;
import com.jwater.core.manager.master.check.WorkerCheck;
import com.jwater.core.manager.master.choose.ChooseHost;
import com.jwater.core.manager.master.choose.RandomChooseHost;
import com.jwater.core.manager.master.event.*;
import com.jwater.core.manager.master.monitor.MonitorManager;
import com.jwater.core.manager.proxy.ProxyClient;
import com.jwater.core.manager.proxy.ProxyManager;
import com.jwater.core.manager.worker.assign.WorkerEnv;
import com.jwater.core.net.Node;
import com.jwater.core.net.NodeHandlerAdapter;
import com.jwater.core.net.group.*;
import com.jwater.core.net.group.data.DurableDataItem;
import com.jwater.core.serialize.ProtoStuffSerialize;
import com.jwater.core.serialize.SerializeUtils;
import com.jwater.launch.HomePath;
import com.jwater.launch.LoaderUtils;
import jwater.org.springframework.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @description: master节点
 * @author: jnner
 * @createDate: 2021/2/5
 * @version: 1.0
 */
public class MasterNodeImpl extends NodeHandlerAdapter implements MasterNode, GroupHandler {
    protected static Logger logger = LoggerFactory.getLogger(MasterNodeImpl.class);
    public static final String DATA_TYPE_EVENT = "event";
    public static final Pattern envPattern = Pattern.compile("\\$\\{(.*?)}");

    // 管理进程编号
    private int id;
    // master组中所有成员
    private List<Member> members;
    // master组节点
    public GroupNode groupNode;
    // master上下文
    private MasterContextImpl masterContext;
    // worker配置解析器
    public WorkerConfParse workerConfParse;
    // job配置解析器
    public JobConfParse jobConfParse;
    // core配置解析器
    volatile public CoreConfParse coreConf;
    // 代理进程client管理器
    volatile public ProxyManager proxyManager;
    // 主机选择策略
    private ChooseHost chooseHost;
    // job调度器
    public Map<String, JobScheduler> jobSchedulers = new ConcurrentHashMap<>();
    // job检查器
    private JobCheck jobCheck;
    // worker检查器
    private WorkerCheck workerCheck;
    // core配置文件改变监听
    private CoreConfCheck coreConfCheck;
    // 上线时间
    private long uptime;
    // 客户端node
    public Map<String, Node> clientNodes = new ConcurrentHashMap<>();
    // 接受通知的客户端
    private Map<String, Node> notifyClientNodes = new ConcurrentHashMap<>();
    // worker环境变量信息
    public Map<String, WorkerEnv> workerEnvs = new ConcurrentHashMap<>();
    // 序列化工具
    private ProtoStuffSerialize protoStuffSerialize = new ProtoStuffSerialize();
    // 监控管理器
    private MonitorManager monitorManager;

    public MasterNodeImpl(int id, List<Member> members, CoreConfParse coreConf) {
        uptime = System.currentTimeMillis();
        this.id = id;
        this.members = members;
        this.coreConf = coreConf;
        masterContext = (MasterContextImpl) MasterFactory.getMasterContext();
        masterContext.setMasterNode(this);
        // 解析worker配置
        workerConfParse = new WorkerConfParse();
        workerConfParse.parse();
        // 解析job配置
        jobConfParse = new JobConfParse();
        jobConfParse.parse();
        // 初始化proxy client
        proxyManager = new ProxyManager(coreConf.getProxyInfos());
        // 初始化主机选择器
        chooseHost = new RandomChooseHost(this);
        // 初始化job检测器
        jobCheck = new JobCheck(this);
        jobCheck.start();
        // 初始化worker检查器
        workerCheck = new WorkerCheck(this);
        workerCheck.start();
        // 初始化core check
        coreConfCheck = new CoreConfCheck(id, this);
        // 监控管理器
        monitorManager = new MonitorManager();
    }

    @Override
    public void start() {
        // 打印所有的jar包信息
        List<String> allJars = LoaderUtils.getAllJars();
        for (String allJar : allJars) {
            logger.info(allJar);
        }
        groupNode = new SimpleGroupNode(members);
        // 设置哪些数据不用持久化存储
        groupNode.setData(DataKeys.KEY_ONLINE, false);
        groupNode.setData(DataKeys.KEY_JOB_LAST_RUN_TIME, false);
        // 设置group通信回调
        groupNode.setGroupHandler(this);
        groupNode.setNodeHandler(this);
        // 加入master的group
        groupNode.joinGroup(id);
        logger.info("master start success");
        // 启动proxy client
        proxyManager.setMasterId(id);
        proxyManager.start(coreConf);
        // 启动core check
        coreConfCheck.start();
    }

    @Override
    public WorkerConfParse getWorkerConfParse() {
        return workerConfParse;
    }

    @Override
    public JobConfParse getJobConfParse() {
        return jobConfParse;
    }

    @Override
    public void dispatchEvent(GroupEvent event) {
        try {
            GroupExecutorUtils.execute(() -> onEvent(event));
            byte[] data = SerializeUtils.serializeArray(event.getClass().getName(), event.serialize());
            groupNode.sendDataToMembers(DATA_TYPE_EVENT, data);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    @Override
    public void storeDataItem(String key, String itemKey, String data) {
        DurableDataItem durableDataItem = new DurableDataItem();
        durableDataItem.setVersion(System.currentTimeMillis());
        durableDataItem.setData(data);
        groupNode.storeDataItem(key, itemKey, durableDataItem);
    }

    @Override
    public void removeDataItem(String key, String itemKey) {
        groupNode.removeDataItem(key, itemKey);
    }

    @Override
    public String getDataItem(String key, String itemKey) {
        DurableDataItem durableDataItem = groupNode.getDataItem(key, itemKey);
        if (durableDataItem != null) {
            return durableDataItem.getData();
        }
        return null;
    }

    @Override
    public WorkerInfo getWorkerInfo(String regionName, String deployName, String groupName, String workerName) {
        return workerConfParse.getWorkerInfo(regionName, deployName, groupName, workerName);
    }

    @Override
    public void addWorker(WorkerInfo workerInfo) {
        AddWorkerEvent event = new AddWorkerEvent();
        event.setWorkerInfo(workerInfo);
        dispatchEvent(event);
    }

    @Override
    public void removeWorker(String regionName, String deployName, String groupName, String workerName) {
        RemoveWorkerEvent event = new RemoveWorkerEvent();
        WorkerInfo workerInfo = new WorkerInfo();
        workerInfo.setRegionName(regionName);
        workerInfo.setDeployName(deployName);
        workerInfo.setGroupName(groupName);
        workerInfo.setWorkerName(workerName);
        event.setWorkerInfo(workerInfo);
        dispatchEvent(event);
    }

    @Override
    public void removeGroupWorker(String regionName, String deployName, String groupName) {
        RemoveGroupWorkerEvent event = new RemoveGroupWorkerEvent();
        WorkerInfo workerInfo = new WorkerInfo();
        workerInfo.setRegionName(regionName);
        workerInfo.setDeployName(deployName);
        workerInfo.setGroupName(groupName);
        event.setWorkerInfo(workerInfo);
        dispatchEvent(event);
    }

    @Override
    public List<WorkerInfo> getWorkerList(String regionName) {
        List<WorkerInfo> result = new ArrayList<>();
        List<WorkerInfo> workerInfos = workerConfParse.getRegionWorkerInfo(regionName);
        for (WorkerInfo workerInfo : workerInfos) {
            String prefix = DataKeys.getWorkerKey(regionName, workerInfo.getDeployName(), workerInfo.getGroupName(),
                    workerInfo.getWorkerName()) + ":";
            Map<String, DurableDataItem> runs = groupNode.getItemsByPrefix(DataKeys.KEY_RUN_HOST, prefix);
            if (runs.isEmpty()) {
                workerInfo.setIndex(1);
                workerInfo.setOnline(false);
                result.add(workerInfo);
            } else {
                for (Entry<String, DurableDataItem> en : runs.entrySet()) {
                    String[] items = en.getKey().split(":");
                    int index = Integer.parseInt(items[4]);
                    WorkerInfo copy = workerInfo.copy();
                    copy.setIndex(index);
                    copy.setHost(en.getValue().getData());
                    NodeInfo nodeInfo = coreConf.getNodeInfo(copy.getHost());
                    if (nodeInfo != null) {
                        copy.setHomePath(HomePath.getRemoteHomePath(nodeInfo.getPath()));
                    }
                    String online = getDataItem(DataKeys.KEY_ONLINE, prefix + index);
                    if (online != null) {
                        copy.setOnline(true);
                        Long startTime = Long.parseLong(online.split(":")[5]);
                        copy.setStartTime(startTime);
                        setUI(copy, prefix + index);
                    }
                    setReportInfo(copy, prefix + index);
                    result.add(copy);
                }
            }
        }
        return result;
    }

    private void setReportInfo(WorkerInfo workerInfo, String key) {
        ReportInfo reportInfo = monitorManager.getReportInfo(key);
        if (reportInfo == null) {
            return;
        }
        workerInfo.setPid(reportInfo.getPid());
        workerInfo.setHeapMemory(reportInfo.getHeapMemory());
        workerInfo.setNonHeapMemory(reportInfo.getNonHeapMemory());
        workerInfo.setLiveThreadCount(reportInfo.getLiveThreadCount());
    }

    private void setUI(WorkerInfo workerInfo, String key) {
        String ui = workerInfo.getProps().getString("ui");
        if (StringUtils.isEmpty(ui)) {
            return;
        }
        WorkerEnv workerEnv = workerEnvs.get(key);
        if (workerEnv == null) {
            return;
        }
        ui = replaceEnv(ui, workerEnv);
        workerInfo.setUi(ui);
    }

    private String replaceEnv(String url, WorkerEnv workerEnv) {
        Map<String, String> map = new HashMap<>();
        Matcher matcher = envPattern.matcher(url);
        while (matcher.find()) {
            String kk = matcher.group();
            String k = kk.replace("${", "").replace("}", "");
            if (k.equals("ip")) {
                map.put(kk, workerEnv.envs().get("$host"));
            } else {
                String v = workerEnv.envs().get(k);
                if (v != null) {
                    map.put(kk, v);
                }
            }
        }
        for (Entry<String, String> en : map.entrySet()) {
            url = url.replace(en.getKey(), en.getValue());
        }
        return url;
    }

    @Override
    public List<String> getWorkerConfList(String regionName, String deployName) {
        List<String> result = new ArrayList<String>();
        List<String> confDirs = getWorkerConfRoot(regionName, deployName);
        for (String confDir : confDirs) {
            File file = new File(HomePath.getRegion(), confDir);
            if (file.exists()) {
                listFiles(result, file, true);
            }
        }
        return result;
    }

    private static void listFiles(List<String> list, File dir, boolean recursive) {
        File[] files = dir.listFiles();
        if (files == null) {
            return;
        }
        for (File file : files) {
            if (file.isDirectory()) {
                if (recursive) {
                    listFiles(list, file, recursive);
                }
            } else {
                String p = file.getPath().replace("\\", "/");
                if (!list.contains(p)) {
                    list.add(p);
                }
            }
        }
    }

    @Override
    public List<String> getWorkerConfRoot(String regionName, String deployName) {
        List<String> confDirs = new ArrayList<String>();
        String prefix = HomePath.getRegion();
        File deployPath = WorkerConfParse.getDeployPath(regionName, deployName);
        confDirs.add(new File(deployPath, "conf").getPath().replace("\\", "/").replace(prefix, "/"));
        List<WorkerInfo> workerInfos = workerConfParse.getDeployWorkerInfo(regionName, deployName);
        for (WorkerInfo workerInfo : workerInfos) {
            String confDir = workerInfo.getProps().getString("confDir");
            if (!StringUtils.isEmpty(confDir)) {
                for (String conf : confDir.split(",")) {
                    String path = new File(deployPath, conf).getPath().replace("\\", "/").replace(prefix, "/");
                    if (!confDirs.contains(path)) {
                        confDirs.add(path);
                    }
                }
            }
            String include = workerInfo.getProps().getString("include");
            if (!StringUtils.isEmpty(include)) {
                for (String otherDeployName : include.split(",")) {
                    File otherDeployPath = WorkerConfParse.getDeployPath(regionName, otherDeployName);
                    String path = new File(otherDeployPath, "conf").getPath().replace("\\", "/").replace(prefix, "/");
                    if (!confDirs.contains(path)) {
                        confDirs.add(path);
                    }
                }
            }
        }
        return confDirs;
    }

    @Override
    public void startWorker(String regionName, String deployName, String groupName, String workerName, int index) {
        WorkerInfo workerInfo = workerConfParse.getWorkerInfo(regionName, deployName, groupName, workerName);
        String itemKey = DataKeys.getRunWorkerKey(regionName, deployName, groupName, workerName, index);
        logger.info("startWorker:" + itemKey);
        if (workerInfo == null) {
            throw new ManagerException("start worker error,can not find worker conf for:" + itemKey);
        }
        String oldHost = getDataItem(DataKeys.KEY_RUN_HOST, itemKey);
        NodeInfo nodeInfo = chooseHost.choose(oldHost, workerInfo, index);
        ProxyClient proxyClient = proxyManager.getProxyClient(nodeInfo.getHost());
        if (proxyClient == null || !proxyClient.isConnect()) {
            throw new ManagerException(
                    "start worker error," + nodeInfo.getHost() + " proxy not online,worker for:" + itemKey);
        }
        String jvmArgs = workerInfo.getProps().getString("jvm", "-Xms128m -Xmx1g");
        proxyClient.startWorker(nodeInfo.getHost(), regionName, deployName, groupName, workerName, index, jvmArgs);
    }

    private Node findNode(String itemKey) {
        for (Node node : clientNodes.values()) {
            String runWorkerKey = (String) node.getAttribute(DataKeys.KEY_RUN_WORKER);
            if (runWorkerKey != null && runWorkerKey.equals(itemKey)) {
                return node;
            }
        }
        return null;
    }

    @Override
    public void stopWorker(String regionName, String deployName, String groupName, String workerName, int index,
                           boolean isForce) {
        String itemKey = DataKeys.getRunWorkerKey(regionName, deployName, groupName, workerName, index);
        String host = getDataItem(DataKeys.KEY_RUN_HOST, itemKey);
        logger.info("stopWorker:" + itemKey);
        if (StringUtils.isEmpty(host)) {
            throw new ManagerException("stop worker error,can not find worker host for:" + itemKey);
        }
        WorkerInfo workerInfo = workerConfParse.getWorkerInfo(regionName, deployName, groupName, workerName);
        notifyExit(itemKey, workerInfo);
        ProxyClient proxyClient = proxyManager.getProxyClient(host);
        if (proxyClient == null || !proxyClient.isConnect()) {
            throw new ManagerException("stop worker error," + host + " proxy not online,worker for:" + itemKey);
        }
        removeDataItem(DataKeys.KEY_RUN, itemKey);
        proxyClient.stopWorker(regionName, deployName, groupName, workerName, index, isForce);
    }

    private void notifyExit(String itemKey, WorkerInfo workerInfo) {
        if (workerInfo == null) {
            return;
        }
        boolean notifyExit = workerInfo.getProps().getString("notifyExit", "false").equals("true");
        if (!notifyExit) {
            return;
        }
        Node clientNode = findNode(itemKey);
        if (clientNode != null) {
            logger.info("notifyExit:" + itemKey);
            clientNode.syncCall(MsgKeys.NOTIFY_EXIT, new byte[0]);
        }
    }

    public void dispatchRepeatWorkerEvent(String itemKey, String host) {
        //发送事件，告知所有master重复登录的worker
        RepeatWorkerEvent event = new RepeatWorkerEvent();
        event.setItemKey(itemKey);
        event.setHost(host);
        this.dispatchEvent(event);
    }

    @Override
    public List<JobInfo> getJobList(String regionName) {
        List<JobInfo> jobInfos = jobConfParse.getRegionJobInfo(regionName);
        for (JobInfo jobInfo : jobInfos) {
            String itemKey = DataKeys.getJobKey(regionName, jobInfo.getDeployName(), jobInfo.getGroupName(),
                    jobInfo.getJobName());
            String run = getDataItem(DataKeys.KEY_JOB_RUN, itemKey);
            String host = getDataItem(DataKeys.KEY_JOB_RUN_HOST, itemKey);
            jobInfo.setHost(host);
            jobInfo.setRun(run == null ? false : true);
            if (run != null) {
                jobInfo.setUptime(Long.parseLong(run));
            }
            if (!StringUtils.isEmpty(host)) {
                NodeInfo nodeInfo = coreConf.getNodeInfo(host);
                if (nodeInfo != null) {
                    jobInfo.setHomePath(HomePath.getRemoteHomePath(nodeInfo.getPath()));
                }
            }
            String lastRunTime = getDataItem(DataKeys.KEY_JOB_LAST_RUN_TIME, itemKey);
            if (!StringUtils.isEmpty(lastRunTime)) {
                jobInfo.setLastRunTime(Long.parseLong(lastRunTime));
            }
        }
        return jobInfos;
    }

    @Override
    public void startJob(String regionName, String deployName, String groupName, String jobName) {
        JobInfo jobInfo = jobConfParse.getJobInfo(regionName, deployName, groupName, jobName);
        if (jobInfo == null) {
            throw new ManagerException(
                    "can not find job for:" + DataKeys.getJobKey(regionName, deployName, groupName, jobName));
        }
        StartJobEvent startJobEvent = new StartJobEvent();
        startJobEvent.setJobInfo(jobInfo);
        dispatchEvent(startJobEvent);
    }

    @Override
    public void stopJob(String regionName, String deployName, String groupName, String jobName) {
        JobInfo jobInfo = jobConfParse.getJobInfo(regionName, deployName, groupName, jobName);
        if (jobInfo == null) {
            throw new ManagerException(
                    "can not find job for:" + DataKeys.getJobKey(regionName, deployName, groupName, jobName));
        }
        StopJobEvent stopJobEvent = new StopJobEvent();
        stopJobEvent.setJobInfo(jobInfo);
        dispatchEvent(stopJobEvent);
    }

    @Override
    public void runJob(String regionName, String deployName, String groupName, String jobName, long time,
                       boolean once) {
        JobInfo jobInfo = jobConfParse.getJobInfo(regionName, deployName, groupName, jobName);
        if (jobInfo == null) {
            throw new ManagerException(
                    "can not find job for:" + DataKeys.getJobKey(regionName, deployName, groupName, jobName));
        }
        logger.info("runJob:" + jobInfo);
        String jobKey = DataKeys.getJobKey(jobInfo.getRegionName(), jobInfo.getDeployName(), jobInfo.getGroupName(),
                jobInfo.getJobName());
        String oldHost = getDataItem(DataKeys.KEY_JOB_RUN_HOST, jobKey);
        NodeInfo nodeInfo = chooseHost.choose(oldHost, jobInfo);
        ProxyClient proxyClient = proxyManager.getProxyClient(nodeInfo.getHost());
        if (proxyClient == null || !proxyClient.isConnect()) {
            throw new ManagerException(
                    "run job error," + nodeInfo.getHost() + " proxy not online,worker for:" + jobKey);
        }
        storeDataItem(DataKeys.KEY_JOB_RUN_HOST, jobKey, nodeInfo.getHost());
        storeDataItem(DataKeys.KEY_JOB_LAST_RUN_TIME, jobKey, System.currentTimeMillis() + "");
        proxyClient.runJob(regionName, deployName, groupName, jobName, time, once);
    }

    @Override
    public void killJob(String regionName, String deployName, String groupName, String jobName) {
        JobInfo jobInfo = jobConfParse.getJobInfo(regionName, deployName, groupName, jobName);
        if (jobInfo == null) {
            throw new ManagerException(
                    "can not find job for:" + DataKeys.getJobKey(regionName, deployName, groupName, jobName));
        }
        logger.info("killJob:" + jobInfo);
        String jobKey = DataKeys.getJobKey(jobInfo.getRegionName(), jobInfo.getDeployName(), jobInfo.getGroupName(),
                jobInfo.getJobName());
        String oldHost = getDataItem(DataKeys.KEY_JOB_RUN_HOST, jobKey);
        ProxyClient proxyClient = proxyManager.getProxyClient(oldHost);
        if (proxyClient == null || !proxyClient.isConnect()) {
            throw new ManagerException("kill job error," + oldHost + " proxy not online,worker for:" + jobKey);
        }
        proxyClient.killJob(regionName, deployName, groupName, jobName);
    }

    @Override
    public void loadConf(String regionName) {
        logger.info("loadConf:" + regionName);
        workerConfParse.parseRegion(regionName);
        jobConfParse.parseRegion(regionName);
        groupNode.callMembers("loadConf", S.toBytes(regionName));
    }

    @Override
    public long getUptime() {
        return uptime;
    }

    @Override
    public List<SystemInfo> getAllSystemInfo() {
        return monitorManager.getAllSystemInfo();
    }

    @Override
    public void removeAlarmInfo(String type) {
        monitorManager.removeAlarmInfo(type);
    }

    @Override
    public List<AlarmInfo> getAllAlarmInfo() {
        return monitorManager.getAllAlarmInfo();
    }

    @Override
    public List<PackageInfo> getPackageInfo() {
        Map<String, PackageInfo> packageInfos = new HashMap<>();
        List<String> regionNames = HomePath.getRegionNames();
        for (String regionName : regionNames) {
            PackageInfo packageInfo = new PackageInfo();
            packageInfo.setRegionName(regionName);
            packageInfo.setPackageCount(HomePath.getDeployNames(regionName).size());
            packageInfos.put(regionName, packageInfo);
        }

        Collection<WorkerInfo> workerInfos = workerConfParse.getWorkerInfos();
        Map<String, List<WorkerInfo>> group = workerInfos.stream().collect(Collectors.groupingBy(WorkerInfo::getRegionName));
        for (Entry<String, List<WorkerInfo>> en : group.entrySet()) {
            PackageInfo packageInfo = packageInfos.get(en.getKey());
            if (packageInfo == null) {
                packageInfo = new PackageInfo();
                packageInfo.setRegionName(en.getKey());
                packageInfos.put(en.getKey(), packageInfo);
            }
            //packageInfo.setWorkerGroupCount(en.getValue().stream().collect(Collectors.groupingBy(WorkerInfo::getGroupName)).size());
            packageInfo.setWorkerCount(en.getValue().stream().collect(Collectors.groupingBy(WorkerInfo::getWorkerName)).size());
            packageInfo.setWorkerOnlineCount(groupNode.getItemsByPrefix(DataKeys.KEY_ONLINE, en.getKey() + ":").size());
        }

        List<JobInfo> jobInfos = jobConfParse.getJobInfos().values().stream().collect(Collectors.toList());
        Map<String, List<JobInfo>> jobGroup = jobInfos.stream().collect(Collectors.groupingBy(JobInfo::getRegionName));
        for (Entry<String, List<JobInfo>> en : jobGroup.entrySet()) {
            PackageInfo packageInfo = packageInfos.get(en.getKey());
            if (packageInfo == null) {
                packageInfo = new PackageInfo();
                packageInfo.setRegionName(en.getKey());
                packageInfos.put(en.getKey(), packageInfo);
            }
            //packageInfo.setJobGroupCount(en.getValue().stream().collect(Collectors.groupingBy(JobInfo::getGroupName)).size());
            packageInfo.setJobCount(en.getValue().stream().collect(Collectors.groupingBy(JobInfo::getJobName)).size());
            packageInfo.setJobRunCount(groupNode.getItemsByPrefix(DataKeys.KEY_JOB_RUN, en.getKey() + ":").size());
        }
        return packageInfos.values().stream().collect(Collectors.toList());
    }

    @Override
    public List<GatewayProxyInfo> getGatewayProxyInfo(String gatewayGroupName, String gatewayWorkerName) {
        return getGatewayProxyInfo(gatewayGroupName, gatewayWorkerName, -1, null, null);
    }

    @Override
    public List<GatewayProxyInfo> getGatewayProxyInfo(String gatewayGroupName, String gatewayWorkerName, int gatewayIndex,
                                                      String gatewayRegionName, String gatewayDeployName) {
        List<GatewayProxyInfo> list = new ArrayList<>();
        List<String> regionNames = HomePath.getRegionNames();
        for (String regionName : regionNames) {
            List<WorkerInfo> workers = getWorkerList(regionName);
            for (WorkerInfo worker : workers) {
                if (!worker.getOnline()) {
                    continue;
                }
                String key = DataKeys.getRunWorkerKey(worker.getRegionName(), worker.getDeployName(), worker.getGroupName(),
                        worker.getWorkerName(), worker.getIndex());
                String gatewayKey = null;
                if (gatewayIndex != -1) {
                    gatewayKey = DataKeys.getRunWorkerKey(gatewayRegionName, gatewayDeployName, gatewayGroupName,
                            gatewayWorkerName, gatewayIndex);
                }
                if (!StringUtils.isEmpty(worker.getProps().getString("httpProxy"))) {
                    GatewayProxyInfo proxyInfo = toHttpProxy(worker, key, gatewayGroupName, gatewayWorkerName, gatewayKey);
                    if (proxyInfo != null) {
                        list.add(proxyInfo);
                    }
                }
                if (!StringUtils.isEmpty(worker.getProps().getString("wsProxy"))) {
                    GatewayProxyInfo proxyInfo = toWsProxy(worker, key, gatewayGroupName, gatewayWorkerName, gatewayKey);
                    if (proxyInfo != null) {
                        list.add(proxyInfo);
                    }
                }
            }
        }
        return list;
    }

    private GatewayProxyInfo toHttpProxy(WorkerInfo worker, String key, String gatewayGroupName, String gatewayWorkerName,
                                         String gatewayKey) {
        try {
            GatewayProxyInfo proxyInfo = new GatewayProxyInfo();
            String httpProxy = worker.getProps().getString("httpProxy");
            String[] items = httpProxy.split("=>");
            if (items.length != 2) {
                logger.warn("http proxy error:" + httpProxy);
            }
            String srcUrl = items[0];
            String destUrl = items[1];
            String gatewayName = getGatewayName(srcUrl);
            if (gatewayName == null) {
                logger.warn("http proxy error:" + httpProxy);
                return null;
            }
            if (!gatewayName.equals(gatewayGroupName + ":" + gatewayWorkerName)) {
                return null;
            }
            String srcPrefix = getPrefix(srcUrl);
            if (StringUtils.isEmpty(srcPrefix)) {
                logger.warn("http proxy error:" + httpProxy);
                return null;
            }
            WorkerEnv workerEnv = workerEnvs.get(key);
            if (workerEnv != null) {
                destUrl = replaceEnv(destUrl, workerEnv);
            }
            proxyInfo.setGatewayGroupName(gatewayGroupName);
            proxyInfo.setGatewayWorkerName(gatewayWorkerName);
            proxyInfo.setType("http");
            proxyInfo.setSrcPrefix(srcPrefix);
            proxyInfo.setDestUrl(destUrl);
            setGatewayNodeInfo(proxyInfo, gatewayKey);
            return proxyInfo;
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
        return null;
    }

    private GatewayProxyInfo toWsProxy(WorkerInfo worker, String key, String gatewayGroupName, String gatewayWorkerName,
                                       String gatewayKey) {
        try {
            GatewayProxyInfo proxyInfo = new GatewayProxyInfo();
            String wsProxy = worker.getProps().getString("wsProxy");
            String[] items = wsProxy.split("=>");
            if (items.length != 2) {
                logger.warn("ws proxy error:" + wsProxy);
            }
            String srcUrl = items[0];
            String destUrl = items[1];
            String gatewayName = getGatewayName(srcUrl);
            if (gatewayName == null) {
                logger.warn("ws proxy error:" + wsProxy);
                return null;
            }
            if (!gatewayName.equals(gatewayGroupName + ":" + gatewayWorkerName)) {
                return null;
            }
            String srcPrefix = getPrefix(srcUrl);
            if (StringUtils.isEmpty(srcPrefix)) {
                logger.warn("ws proxy error:" + wsProxy);
                return null;
            }
            WorkerEnv workerEnv = workerEnvs.get(key);
            if (workerEnv != null) {
                destUrl = replaceEnv(destUrl, workerEnv);
            }
            proxyInfo.setGatewayGroupName(gatewayGroupName);
            proxyInfo.setGatewayWorkerName(gatewayWorkerName);
            proxyInfo.setType("ws");
            proxyInfo.setSrcPrefix(srcPrefix);
            proxyInfo.setDestUrl(destUrl);
            setGatewayNodeInfo(proxyInfo, gatewayKey);
            return proxyInfo;
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
        return null;
    }

    private List<String> getEnvPatternKey(String url) {
        List<String> list = new ArrayList<>();
        Matcher matcher = envPattern.matcher(url);
        while (matcher.find()) {
            String kk = matcher.group();
            String k = kk.replace("${", "").replace("}", "");
            list.add(k);
        }
        return list;
    }

    private String getGatewayName(String url) {
        List<String> keys = getEnvPatternKey(url);
        if (keys.size() == 1) {
            return keys.get(0);
        } else if (keys.size() > 1) {
            for (String key : keys) {
                if (key.indexOf("gateway") > -1) {
                    return key;
                }
            }
            return keys.get(0);
        }
        return null;
    }

    private String getPrefix(String url) {
        int index = url.lastIndexOf("}");
        if (index > -1) {
            return url.substring(index + 1);
        }
        return "";
    }

    private void setGatewayNodeInfo(GatewayProxyInfo proxyInfo, String gatewayKey) {
        if (StringUtils.isEmpty(gatewayKey)) {
            return;
        }
        WorkerEnv workerEnv = workerEnvs.get(gatewayKey);
        if (workerEnv == null) {
            return;
        }
        String host = workerEnv.envs().get("$host");
        String port = workerEnv.envs().get("port");
        if (StringUtils.isEmpty(host) || StringUtils.isEmpty(port)) {
            return;
        }
        proxyInfo.setGatewayHost(host);
        proxyInfo.setGatewayPort(port);
    }

    public void onEvent(GroupEvent event) {
        //logger.info("event," + event.getType() + ":" + JSON.toJSONString(event));
        if (GroupEvent.TYPE_LOGIN.equals(event.getType())) {
            doForLogin((LoginEvent) event);
        } else if (GroupEvent.TYPE_OFFLINE.equals(event.getType())) {
            doForOffline((OfflineEvent) event);
        } else if (GroupEvent.TYPE_ADD_WORKER.equals(event.getType())) {
            AddWorkerEvent addWorkerEvent = (AddWorkerEvent) event;
            workerConfParse.addWorkerInfo(addWorkerEvent.getWorkerInfo());
        } else if (GroupEvent.TYPE_REMOVE_WORKER.equals(event.getType())) {
            RemoveWorkerEvent removeWorkerEvent = (RemoveWorkerEvent) event;
            workerConfParse.removeWorkerInfo(removeWorkerEvent.getWorkerInfo());
        } else if (GroupEvent.TYPE_REMOVE_GROUP_WORKER.equals(event.getType())) {
            RemoveGroupWorkerEvent removeGroupWorkerEvent = (RemoveGroupWorkerEvent) event;
            workerConfParse.removeGroupWorker(removeGroupWorkerEvent.getWorkerInfo());
        } else if (GroupEvent.TYPE_START_JOB.equals(event.getType())) {
            JobInfo jobInfo = ((StartJobEvent) event).getJobInfo();
            doForStartJob(jobInfo);
        } else if (GroupEvent.TYPE_STOP_JOB.equals(event.getType())) {
            JobInfo jobInfo = ((StopJobEvent) event).getJobInfo();
            doForStopJob(jobInfo);
        } else if (GroupEvent.TYPE_COMPLETE_JOB.equals(event.getType())) {
            JobInfo jobInfo = ((CompleteJobEvent) event).getJobInfo();
            doForCompleteJob(jobInfo);
        } else if (GroupEvent.TYPE_REPORT.equals(event.getType())) {
            doForReport((ReportInfoEvent) event);
        } else if (GroupEvent.TYPE_REPEAT_WORKER.equals(event.getType())) {
            doForRepeatWorker((RepeatWorkerEvent) event);
        } else if (GroupEvent.TYPE_PROXY_REPORT.equals(event.getType())) {
            doForProxyReport((ProxyReportInfoEvent) event);
        } else if (GroupEvent.TYPE_ALARM_REPORT.equals(event.getType())) {
            doForAlarmReport((AlarmReportInfoEvent) event);
        }
    }

    private void doForLogin(LoginEvent event) {
        WorkerEnv workerEnv = WorkerEnv.wrap(event.getEnvsString());
        workerEnv.put("$host", event.getHost());
        String itemKey = DataKeys.getRunWorkerKey(event.getRegionName(), event.getDeployName(), event.getGroupName(),
                event.getWorkerName(), event.getIndex());
        workerEnvs.put(itemKey, workerEnv);
        for (Node node : notifyClientNodes.values()) {
            node.sendData(MsgKeys.NOTIFY_WORKER_CHANGE, S.toBytes(itemKey));
        }
    }

    private void doForOffline(OfflineEvent event) {
        String key = DataKeys.getRunWorkerKey(event.getRegionName(), event.getDeployName(), event.getGroupName(),
                event.getWorkerName(), event.getIndex());
        workerEnvs.remove(key);
        if (groupNode.isLeader()) {
            workerCheck.remove(key);
        }
        for (Node node : notifyClientNodes.values()) {
            node.sendData(MsgKeys.NOTIFY_WORKER_CHANGE, S.toBytes(key));
        }
    }

    private void doForRepeatWorker(RepeatWorkerEvent event) {
        for (Node node : clientNodes.values()) {
            String runWorkerKey = (String) node.getAttribute(DataKeys.KEY_RUN_WORKER);
            String host = (String) node.getAttribute(DataKeys.KEY_RUN_HOST);
            if (runWorkerKey != null && runWorkerKey.equals(event.getItemKey()) && event.getHost().equals(host)) {
                node.removeAttribute(runWorkerKey);
                logger.info("node remove attribute:" + runWorkerKey);
            }
        }
        if (groupNode.isLeader()) {
            if (getDataItem(DataKeys.KEY_ONLINE, event.getItemKey()) != null) {
                stopRepeatWorker(event.getItemKey(), event.getHost());
            }
        }
    }

    private void stopRepeatWorker(String itemKey, String host) {
        logger.info("stopRepeatWorker:" + itemKey + "," + host);
        ProxyClient proxyClient = proxyManager.getProxyClient(host);
        if (proxyClient == null || !proxyClient.isConnect()) {
            throw new ManagerException("stop repeat worker error," + host + " proxy not online,worker for:" + itemKey);
        }
        String[] items = itemKey.split(":");
        proxyClient.stopWorker(items[0], items[1], items[2], items[3], Integer.parseInt(items[4]), true);
    }

    private void doForReport(ReportInfoEvent event) {
        ReportInfo reportInfo = protoStuffSerialize.deserialize(event.getData(), ReportInfo.class);
        workerCheck.check(reportInfo);
        monitorManager.addReportInfo(reportInfo);
    }

    private void doForProxyReport(ProxyReportInfoEvent event) {
        SystemInfo systemInfo = protoStuffSerialize.deserialize(event.getData(), SystemInfo.class);
        monitorManager.addSystemInfo(systemInfo);
    }

    private void doForAlarmReport(AlarmReportInfoEvent event) {
        AlarmInfo alarmInfo = protoStuffSerialize.deserialize(event.getData(), AlarmInfo.class);
        monitorManager.addAlarmInfo(alarmInfo);
    }

    private void doForStartJob(JobInfo jobInfo) {
        if (groupNode.isLeader()) {
            logger.info("doForStartJob:" + jobInfo);
            String jobKey = DataKeys.getJobKey(jobInfo.getRegionName(), jobInfo.getDeployName(), jobInfo.getGroupName(),
                    jobInfo.getJobName());
            String itemKey = DataKeys.getJobGroupKey(jobInfo.getRegionName(), jobInfo.getDeployName(),
                    jobInfo.getGroupName());
            JobScheduler jobScheduler = jobSchedulers.get(itemKey);
            if (jobScheduler == null) {
                jobScheduler = new JobScheduler(jobInfo, jobConfParse, this);
                jobSchedulers.put(itemKey, jobScheduler);
            }
            storeDataItem(DataKeys.KEY_JOB_RUN, jobKey, System.currentTimeMillis() + "");
            jobScheduler.startScheduler();
        }
    }

    private void doForStopJob(JobInfo jobInfo) {
        if (groupNode.isLeader()) {
            logger.info("doForStopJob:" + jobInfo);
            String jobKey = DataKeys.getJobKey(jobInfo.getRegionName(), jobInfo.getDeployName(), jobInfo.getGroupName(),
                    jobInfo.getJobName());
            String itemKey = DataKeys.getJobGroupKey(jobInfo.getRegionName(), jobInfo.getDeployName(),
                    jobInfo.getGroupName());
            removeDataItem(DataKeys.KEY_JOB_RUN, jobKey);
            JobScheduler jobScheduler = jobSchedulers.get(itemKey);
            if (jobScheduler != null) {
                if (jobScheduler.tryStopScheduler()) {
                    jobSchedulers.remove(itemKey);
                }
            }
        }
    }

    private void doForCompleteJob(JobInfo jobInfo) {
        if (groupNode.isLeader()) {
            logger.info("doForCompleteJob:" + jobInfo);
            String itemKey = DataKeys.getJobGroupKey(jobInfo.getRegionName(), jobInfo.getDeployName(),
                    jobInfo.getGroupName());
            JobScheduler jobScheduler = jobSchedulers.get(itemKey);
            if (jobScheduler == null) {
                throw new ManagerException("doForCompleteJob error,can not find jobScheduler for:" + itemKey);
            }
            jobScheduler.runNextJob(jobInfo, jobInfo.getTime());
        }
    }

    @Override
    public void onJoin(int memberId, Node node) {
        logger.info("onJoin:" + memberId);
        jobCheck.check();
    }

    @Override
    public void onLeave(int memberId, Node node) {
        logger.info("onLeave:" + memberId);
        jobCheck.check();
    }

    @Override
    public void onMemberData(int memberId, Node node, String dataType, byte[] data) {
        try {
            if (DATA_TYPE_EVENT.equals(dataType)) {
                Object[] objs = SerializeUtils.deserializeArray(data);
                String className = (String) objs[0];
                GroupEvent event = (GroupEvent) Class.forName(className).newInstance();
                event = event.deserialize((byte[]) objs[1]);
                onEvent(event);
            }
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    @Override
    public byte[] onMemberCall(int memberId, Node node, String method, byte[] data) {
        if ("loadConf".equals(method)) {
            String regionName = S.newString(data);
            workerConfParse.parseRegion(regionName);
            jobConfParse.parseRegion(regionName);
        }
        return null;
    }

    @Override
    public void onDataItemEvent(DataItemEvent event) {
        //logger.info("onDataItemEvent," + event);
    }

    @Override
    public void onConnect(Node node) {
        logger.info("onConnect:" + node.getNodeId());
        clientNodes.put(node.getNodeId(), node);
    }

    @Override
    public void onDisconnect(Node node) {
        logger.info("onDisconnect:" + node.getNodeId());
        clientNodes.remove(node.getNodeId());
        notifyClientNodes.remove(node.getNodeId());
        String runWorkerKey = (String) node.getAttribute(DataKeys.KEY_RUN_WORKER);
        if (runWorkerKey != null) {
            // 移除在线状态
            removeDataItem(DataKeys.KEY_ONLINE, runWorkerKey);
            // 获取属性
            String regionName = (String) node.getAttribute(DataKeys.KEY_REGION_NAME);
            String deployName = (String) node.getAttribute(DataKeys.KEY_DEPLOY_NAME);
            String groupName = (String) node.getAttribute(DataKeys.KEY_GROUP_NAME);
            String workerName = (String) node.getAttribute(DataKeys.KEY_WORKER_NAME);
            Integer index = (Integer) node.getAttribute(DataKeys.KEY_INDEX);
            OfflineEvent offlineEvent = new OfflineEvent();
            offlineEvent.setRegionName(regionName);
            offlineEvent.setDeployName(deployName);
            offlineEvent.setGroupName(groupName);
            offlineEvent.setWorkerName(workerName);
            offlineEvent.setIndex(index);
            // 发送离线事件
            dispatchEvent(offlineEvent);
        }
    }

    @Override
    public void onData(Node node, String dataType, byte[] data) {
        // logger.info("onData:" + dataType);
        if (MsgKeys.REPORT.equals(dataType)) {
            ReportInfoEvent event = new ReportInfoEvent();
            event.setData(data);
            this.dispatchEvent(event);
        }
    }

    @Override
    public byte[] onSyncCall(Node node, String method, byte[] args) {
        if (MsgKeys.LOGIN.equals(method)) {
            return doLogin(node, method, args);
        } else if (MsgKeys.GET_UPTIME.equals(method)) {
            return S.toBytes(uptime + "");
        } else if (MsgKeys.ACCEPT_NOTIFY.equals(method)) {
            notifyClientNodes.put(node.getNodeId(), node);
        }
        return new byte[0];
    }

    public byte[] doLogin(Node node, String method, byte[] args) {
        Object[] datas = SerializeUtils.deserializeArray(args);
        String host = (String) datas[0];
        String regionName = (String) datas[1];
        String deployName = (String) datas[2];
        String groupName = (String) datas[3];
        String workerName = (String) datas[4];
        int index = (int) datas[5];
        long startTime = (long) datas[6];
        int pid = (int) datas[7];
        String envsString = (String) datas[8];
        //logger.info("envsString:" + envsString);

        String itemKey = DataKeys.getRunWorkerKey(regionName, deployName, groupName, workerName, index);
        String online = getDataItem(DataKeys.KEY_ONLINE, itemKey);
        if (online != null) {
            // 当一台master挂掉，就会收到该台master机器上的worker再次登录
            logger.info("worker online,again login for:" + itemKey);
            // 以下是判定有2个以上进程重复登录情况
            int oldPid = Integer.parseInt(online.split(":")[6]);
            if (oldPid != pid) {
                logger.info("worker online,repeat login for:" + itemKey);
                return S.toBytes("false");
            }
        }
        // 设置属性
        node.setAttribute(DataKeys.KEY_RUN_WORKER, itemKey);
        node.setAttribute(DataKeys.KEY_REGION_NAME, regionName);
        node.setAttribute(DataKeys.KEY_DEPLOY_NAME, deployName);
        node.setAttribute(DataKeys.KEY_GROUP_NAME, groupName);
        node.setAttribute(DataKeys.KEY_WORKER_NAME, workerName);
        node.setAttribute(DataKeys.KEY_INDEX, index);
        node.setAttribute(DataKeys.KEY_RUN_HOST, host);
        // 存储在线状态
        storeDataItem(DataKeys.KEY_ONLINE, itemKey, itemKey + ":" + startTime + ":" + pid);
        // 标注为运行状态
        storeDataItem(DataKeys.KEY_RUN, itemKey, "true");
        storeDataItem(DataKeys.KEY_RUN_HOST, itemKey, host);
        // 发送在线事件
        LoginEvent loginEvent = new LoginEvent();
        loginEvent.setRegionName(regionName);
        loginEvent.setDeployName(deployName);
        loginEvent.setGroupName(groupName);
        loginEvent.setWorkerName(workerName);
        loginEvent.setIndex(index);
        loginEvent.setLoginTime(System.currentTimeMillis());
        loginEvent.setStartTime(startTime);
        loginEvent.setHost(host);
        loginEvent.setEnvsString(envsString);
        dispatchEvent(loginEvent);
        return S.toBytes("true");
    }
}
